package cn.demo.rocketmq.transaction.demo1;

import cn.demo.rocketmq.transaction.demo1.util.Account;
import cn.demo.rocketmq.transaction.demo1.util.Journal;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
//import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;

/**
 * @author: WangSaiChao
 * @date: 2018/2/12
 * @description: 消息消费者
 */
@Component
public class MessageListenerImpl implements MessageListenerConcurrently {


    @Autowired
    private AccountService accountService;


    @Override
//    @Transactional
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt msg = msgs.get(0);
        try {

            String topic = msg.getTopic();
            //Message Body
            JSONObject messageBody = JSONObject.parseObject(new String(msg.getBody(), "UTF-8"));
            String tags = msg.getTags();
            String keys = msg.getKeys();

            String userId = messageBody.getString("userId");
            String orderId = messageBody.getString("orderId");
            Double amount = messageBody.getDouble("amount");

            //打印
            System.out.println(topic + "  " + tags + "    " + keys + "  用户id为： " + userId + ",订单号为： " + orderId + ",消费金额为：" + amount);

            //业务逻辑处理,根据用户id去数据库中 将账户余额更新
            Account account = new Account();
            Date date = new Date();
            account.setAmount(amount);
            account.setUserId(userId);
            account.setUpdateTime(date);
            accountService.update(account);

            //保存记录到账户流水表
            Journal journal = new Journal();
            journal.setOrderId(orderId);
            journal.setUserId(userId);
            journal.setAmount(amount);
            journal.setUpdateTime(date);
            accountService.insert(journal);

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

        } catch (Exception e) {
            e.printStackTrace();
            //重试次数为3情况
            if (msg.getReconsumeTimes() == 3) {

                /**
                 * 如果连续重发3次还是发布出去 证明该订单可能哪里有错
                 * 保存错误日志,然后就不再发送消息了,需要人工介入,逻辑我就不写了
                 */
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            //如果没超过三次,稍后再次发送
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }

    }
}
